package strom_kafka.topology;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.AuthorizationException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.ZkHosts;
import strom_kafka.bolt.WordSpliter;
import strom_kafka.bolt.WriterBolt;
import strom_kafka.spout.MessageScheme;

import java.util.Arrays;

/**
 * @auther Fighter Created on 2018/4/9.
 */
public class KafkaTopo {
    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
        String topic = "test";
        String zkRoot = "/storm-kafka-test";
        String spoutID = "kafkaSpout";
        BrokerHosts zkHosts = new ZkHosts("192.168.118.133:2181");

        SpoutConfig spoutConfig = new SpoutConfig(zkHosts, topic, zkRoot, spoutID);
        spoutConfig.scheme = new SchemeAsMultiScheme(new MessageScheme());
        spoutConfig.zkServers= Arrays.asList(new String[]{"192.168.118.133"});
        spoutConfig.zkPort = 2181;
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout(spoutID, new KafkaSpout(spoutConfig));
        builder.setBolt("word-spliter", new WordSpliter()).shuffleGrouping(spoutID);
        builder.setBolt("writer", new WriterBolt(), 1).fieldsGrouping("word-spliter", new Fields("word"));
        Config config = new Config();
        config.setNumWorkers(4);
        config.setNumAckers(0);
        config.setDebug(false);

        //提交本地
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("wordCount", config, builder.createTopology());
        //提交集群
//        StormSubmitter.submitTopology("wordCount",config,builder.createTopology());

    }
}
